from Config.myConstant import *
from Config.myConfig import *
import datetime
from DataAccess.TradedayDataProcess import *
import pymssql
import dateutil.parser as dtparser
from joblib import Parallel, delayed,parallel_backend
from Utility.InfluxdbUtility import InfluxdbUtility 
import influxdb
import warnings

########################################################################
class MinuteKLineProcess(object):
    """description of class"""
    #----------------------------------------------------------------------
    def __init__(self,SqlSource=SqlServer['server170'],OracleSource=OracleServer['default']):
        strArry=SqlSource.split(';')
        self.address=strArry[0].split('=')[1]
        self.user=strArry[1].split('=')[1]
        self.password=strArry[2].split('=')[1]
        self.oracleConnectStr=OracleSource
        #self.influxdb=self.__connectInfluxdb(InfluxdbServer)
        #database='MinuteLine'
        #self.sqlserver=pymssql.connect( self.address,self.user,self.password,database,charset='utf8')
    #----------------------------------------------------------------------
    def __connectInfluxdb(self,InfluxdbSource):
        host=InfluxdbSource['host']
        port=InfluxdbSource['port']
        username=InfluxdbSource['username']
        password=InfluxdbSource['password']
        database=InfluxdbSource['database']
        client = influxdb.DataFrameClient(host=host, port=port, username=username, password=password, database=database)
        return client
        pass
    #----------------------------------------------------------------------
    def __getCodeListFromSql(self):
        database='MinuteLine'
        connect=pymssql.connect( self.address,self.user,self.password,database,charset='utf8')
        cursor = connect.cursor()
        sql="SELECT Name FROM SysObjects Where XType='U' ORDER BY Name"
        cursor.execute(sql)
        mydata=cursor.fetchall()
        codeList=[]
        for table in mydata:
            name=table[0]
            if 'Min' in name:
                code=name.split('_')[1]+'.'+name.split('_')[2]
                if (((code[0]=='0') | (code[0]=='6') | (code[0]=='3') |(code[0]=='5') |((code[0]=='1')& (code[1]=='1'))|((code[0]=='1')& (code[1]=='2'))) &(len(name.split('_')[1])==6)) |((code[0]=='1') &(len(name.split('_')[1])==8)) :
                    codeList.append(code)
        return codeList

        pass
    #----------------------------------------------------------------------
    def __getAllStockIPOInfoFromRDF(self):
        oracleConnectString=OracleServer['default']
        myConnection=oracle.connect(oracleConnectString)
        myCursor=myConnection.cursor()
        oracleStr="select s_info_windcode,S_INFO_NAME,S_INFO_EXCHMARKET,S_INFO_LISTBOARD,S_INFO_LISTDATE,S_INFO_DELISTDATE from wind_filesync.AShareDescription order by S_INFO_LISTDATE"
        myCursor.execute(oracleStr)
        mydata=pd.DataFrame(myCursor.fetchall(),columns=['code','name','exchange','board','listDate','delistDate'])
        return mydata
        pass
    #----------------------------------------------------------------------
    #输入code=600000.SH，startdate=yyyyMMdd，endDate=yyyyMMdd
    def __getMinuteDataByDateFromSqlSever(self,code,startDate=EMPTY_STRING,endDate=EMPTY_STRING):
        database='MinuteLine'
        connect=pymssql.connect( self.address,self.user,self.password,database,charset='utf8')
        #connect=self.sqlserver
        cursor = connect.cursor()
        if startDate==EMPTY_STRING:
                sql = "SELECT [stkcd] as [code], [tdate] as [date],[ttime] as [minute],[Open] as [open] ,[High] as [high],[Low] as [low],[Close] as [close] ,[Volume] as [volume],[Amount] as [amount]  FROM [{0}].[dbo].[Min1_{1}] order by [tdate],[ttime]".format(database,code.replace('.','_'))
        elif endDate==EMPTY_STRING:
            sql = "SELECT [stkcd] as [code], [tdate] as [date],[ttime] as [minute],[Open] as [open] ,[High] as [high],[Low] as [low],[Close] as [close] ,[Volume] as [volume],[Amount] as [amount]  FROM [{0}].[dbo].[Min1_{1}] where [tdate]>={2} order by [tdate],[ttime]".format(database,code.replace('.','_'),startDate)
        else:
            sql = "SELECT [stkcd] as [code], [tdate] as [date],[ttime] as [minute],[Open] as [open] ,[High] as [high],[Low] as [low],[Close] as [close] ,[Volume] as [volume],[Amount] as [amount]  FROM [{0}].[dbo].[Min1_{1}] where [tdate]>={2} and [tdate]<={3} order by [tdate],[ttime]".format(database,code.replace('.','_'),startDate,endDate)
            
        cursor.execute(sql)
        mydata=cursor.fetchall()
        if (len(mydata)>0):
            mydata = pd.DataFrame(mydata,columns=['stkcd' ,'tdate','ttime' ,'Open' ,'High','Low','Close' ,'Volume' ,'Amount'])
            mydata['mytime']=pd.to_datetime(mydata['tdate']+mydata['ttime'],format='%Y%m%d%H%M')
            mydata.set_index('mytime',inplace=True,drop=True)
            mydata[['Open' ,'High','Low','Close' ,'Volume' ,'Amount']] = mydata[['Open' ,'High','Low','Close' ,'Volume' ,'Amount']].astype('float')
        else:
            mydata=pd.DataFrame()
        return mydata   
    #----------------------------------------------------------------------
    def getMinuteDataFromSqlServerByCode(self,code,startDate,endDate):
        return self.__getMinuteDataByDateFromSqlSever(code,startDate,endDate)
        pass
    #----------------------------------------------------------------------
    def getMinuteDataFromInfluxdbByCode(self,code,startDate,endDate):
        database=INFLUXDBMINUTEDATABASE
        measurement=code
        dateStart=str(startDate)
        dateEnd=str(endDate)
        begin = dtparser.parse(dateStart)
        end = dtparser.parse(dateEnd)
        b = begin + datetime.timedelta(hours=0)
        e = end + datetime.timedelta(hours=24)
        query = f""" select * from "{database}"."autogen"."{measurement}" where time >= {int(b.timestamp() * 1000 * 1000 * 1000)} and time < {int(e.timestamp() * 1000 * 1000* 1000)} """
        connect=self.__connectInfluxdb(InfluxdbServer)
        result=connect.query(query)
        data=pd.DataFrame()
        try:
            data=result[measurement]
            data[['Amount', 'Close', 'High', 'Low', 'Open', 'Volume']] = data[['Amount', 'Close', 'High', 'Low', 'Open','Volume']].astype('float')
        except:
            pass

        return data
        pass
    #----------------------------------------------------------------------
    def getMinuteDataFromInfluxdbByCodeList(self,codeList,startDate,endDate):
        data=Parallel(n_jobs=MYJOBS,verbose=0)(delayed(self.getMinuteDataFromInfluxdbByCode)(code,startDate,endDate) for code in codeList)
        data=pd.concat(data)
        return data
        pass
    #----------------------------------------------------------------------
    def updateMinuteKineToInfluxdbByCode(self,code,startDate=EMPTY_STRING,endDate=EMPTY_STRING):
        #allCodes=InfluxdbUtility.getAllMeausrements('MinuteLine')
        #logger.info(f'save minuteData of {code} to Influxdb start!!! ')
        try:
            data=self.__getMinuteDataByDateFromSqlSever(code,startDate,endDate)
            if (data.shape[0]>0):
                length=data.shape[0]
                for i in range(0,length,10000):
                    data0=data.iloc[i*10000:min((i+1)*10000,length)]
                    InfluxdbUtility.saveDataFrameDataToInfluxdb(data0,INFLUXDBMINUTEDATABASE,str(code),{},InfluxdbServerLocal)
                logger.info(f'save minuteData of {code} to Influxdb finished!!! ')
        except:
            logger.info(f"minuteData of {code} occurs error!!")
    #----------------------------------------------------------------------
    def getAllMeasurement():
        
        pass
    #----------------------------------------------------------------------
    def updateMinuteKineToInfluxdb(self,startDate=EMPTY_STRING,endDate=EMPTY_STRING):
        #mylist=self.__getCodeListFromSql()
        ipoData=self.__getAllStockIPOInfoFromRDF()
        ipoData=ipoData[ipoData['listDate'].isna()==False]
        codeList=list(ipoData['code'])
        warnings.filterwarnings('ignore')
        Parallel(n_jobs=MYJOBS,verbose=0)(delayed(self.updateMinuteKineToInfluxdbByCode)(code,startDate,endDate) for code in codeList)
        pass
########################################################################